package com.xushuda.yas.storage;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

import com.xushuda.yas.storage.OpLog.DMOP;

/**
 * also refer to double slice table LSMTable,MemSlice,PermSlice合起来是一个LSMTree的简易实现
 * 
 * @author xushuda
 *
 */
public class LSMTable implements Iterable<String> {

    /**
     * perm和log日志的根目录
     */
    public static final String BASE_DIR = "/opt/yas";

    /**
     * mutable map in memory
     */
    private MemSlice memSlice;
    /**
     * the immutable map in memory \ indicates : in memory and immutable
     */
    private volatile MemSlice immSlice;

    /**
     * persistence on disk
     */
    private PermSlice permSlice;

    /**
     * in case of multiple compacting thread
     */
    private volatile boolean compacting = false;

    /**
     * comparator of the key
     */
    private Comparator<String> comparator;

    /**
     * 日志类，记录日志 日志也用于恢复
     */
    private OpLog opLog;

    @Deprecated
    public LSMTable() throws IOException {

        // 创建根目录
        File baseDir = new File(BASE_DIR);
        if (!baseDir.exists()) {
            if (!baseDir.mkdirs()) {
                throw new IOException("create base directories fail");
            }
        }

        String logDir = BASE_DIR;
        String permDir = BASE_DIR + "/data";

        // 默认按lexical字典顺序排序
        comparator = new Comparator<String>() {

            @Override
            public int compare(String o1, String o2) {
                return o1.compareTo(o2);
            }
        };

        // 参数为cmp和数据文件根目录
        permSlice = SliceFactory.getPermSliceInstance(comparator, permDir);

        // 参数为具体文件路径
        opLog = new OpLog(logDir);

        // 重建内存表
        reconstruct();
    }

    /**
     * 根据opLog中的数据，和perm中最大的序列号对比，重建内存表部分
     * 
     */
    private void reconstruct() {

        // TODO
        memSlice = SliceFactory.getMemSliceInstance(comparator, false);
        immSlice = null;
    }

    /**
     * compact memory table with data on disk 这个调用会开启新的线程不会阻塞当前线程
     */
    private void compact() {
        if (compacting) {
            return;
        }
        // -------compacting start----------

        compacting = true;
        // this operation should not be in other thread in case of concurrency problem
        immSlice = memSlice.flushMap();
        // flushMap的调用是同步的，将数据合并到磁盘数据的过程是开启新的进程异步运行的
        new Thread(new Runnable() {

            @Override
            public void run() {
                // permSlice负责把immSlice中的数据dump到磁盘上

                try {
                    permSlice.dump(immSlice.range(null, null));
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                // immSlice设为null
                immSlice = null;
                compacting = false;
                // -------compacting end----------
            }
        }).start();
    }

    public String update(String key, String value) {
        return null;
    }

    /**
     * insert the data
     * 
     * @param key
     * @param value
     * @return if any existing key matches param key return the value else return null
     * 
     */
    public String insert(String key, String value) {
        // if (memSlice.contains(key)) {
        // return memSlice.get(key);
        // } else if (permSlice.contains(key)) {
        // return permSlice.get(key);
        // } else if (null != immSlice && immSlice.contains(key)) {
        // return immSlice.get(key);
        // }

        if (memSlice.contains(key, opLog.getSeq())) {
            return null;
        }
        // insert into the memSlice
        try {
            long seqNo = opLog.writeLog(key, value, DMOP.INSERTION);
            memSlice.insert(key, value, seqNo);
        } catch (IOException e) {
            e.printStackTrace();
        }

        // check flush
        if (memSlice.shouldFlush()) {
            compact();
        }

        return null;
    }

    /**
     * get the data from existing file system(memory and disk) 已经加入了mvcc版本控制
     * 
     * @param key
     * @return
     */
    public String get(String key) {

        long currentSeqNo = opLog.getSeq();
        String str;
        if (null != (str = memSlice.find(key, currentSeqNo))
                || (null != immSlice && null != (str = immSlice.find(key, currentSeqNo)))
                || null != (str = permSlice.find(key, currentSeqNo))) {
            return str;
        }
        return null;
    }

    /**
     * 获取在区间范围内的所有数据
     * 
     * @param startKey
     * @param endKey
     * @return
     */
    public String[] slice(String startKey, String endKey) {
        return null;
    }

    /**
     * 获取大于startKey的所有数据
     * 
     * @param startKey
     * @return
     */
    public String[] sliceUp(String startKey) {
        return null;
    }

    /**
     * 获取小于endKey的所有数据
     * 
     * @param endKey
     * @return
     */
    public String[] sliceDown(String endKey) {
        return null;
    }

    /**
     * 反向迭代器
     * 
     * @param startKey
     * @return
     */
    public Iterator<String> iteratorReverse(String startKey) {
        return new InnerIterator(startKey);
    }

    /**
     * 正向迭代器
     */
    @Override
    public Iterator<String> iterator() {
        return new InnerIterator();
    }

    // /**
    // * position for innerIterator
    // *
    // * @author xushuda
    // *
    // */
    // static enum LEASTPOS {
    // PermK(0), MEMK(1), IMMK(2);
    // public int value;
    //
    // private LEASTPOS(int value) {
    // this.value = value;
    // }
    // }

    /**
     * Iterator : 每个Iterator本身需要保持immutable Table的副本，配合版本控制实现mvcc TODO 迭代器实现
     * 
     * @author xushuda
     *
     */
    @SuppressWarnings("unused")
    private class InnerIterator implements Iterator<String> {

        String currentKey;

        // 获取这个Iterator的时候的序列号
        private long seqNo;

        private String permK;
        private String memK;
        private String immK;

        public InnerIterator(String currentKey) {
            this.currentKey = currentKey;
        }

        /**
         * status changed while the imMap change notice that if your process on one iterator is too long there might
         * still be a problem
         */
        private void checkNreload() {
            // imMap has changed
            init(currentKey);
        }

        private void init(String key) {

        }

        // private String getLeast(String permK, String memK, String immK) {
        // comparator.compare(permK, memK);
        // }

        public InnerIterator() {
            init(null);
            // currentKey = getLeast(permK, memK, immK);
        }

        @Override
        public boolean hasNext() {
            checkNreload();
            // TODO Auto-generated method stub
            return false;
        }

        @Override
        public String next() {
            // TODO Auto-generated method stub
            return null;
        }

        @Override
        public void remove() {
            // TODO Auto-generated method stub

        }

    }

    public static void main(String[] args) throws IOException {
        LSMTable ls = new LSMTable();
        // 数据量 100w
        int num = 1 << 20;
        // 生成数据dada 中为顺序的数据
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < 128; i++) {
            sb.append("xx");
        }
        String x = sb.toString();
        List<String> data = new ArrayList<String>();
        for (int i = num; i > 0; i--) {
            data.add(x + i);
        }

        System.out.println("开始写入测试");
        String y = sb.append("dsds").toString();
        long st = System.currentTimeMillis();
        for (String tmp : data) {
            ls.insert(tmp, y);
        }
        System.out.println((System.currentTimeMillis() - st) + "毫秒");

        System.out.println("开始顺序读取测试");
        long st2 = System.currentTimeMillis();
        for (String tmp : data) {
            ls.get(tmp);
        }
        System.out.println((System.currentTimeMillis() - st2) + "毫秒");

        java.util.Collections.shuffle(data);
        System.out.println("开始随机读取测试");
        // 大量随机query，记录qps
        long st3 = System.currentTimeMillis();
        for (String tmp : data) {
            ls.get(tmp);
        }
        System.out.println((System.currentTimeMillis() - st3) + "毫秒");

    }
}
